package com.ken.app.kafka.producer;

import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.inject.Inject;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @Author: Liuk
 * @Mail: leo_work@aliyun.com
 * @Date: 2017/8/18 0018
 * @Desc: kafka异步发送封装
 */
@Component
@Slf4j
public class KafkaSenderService {

    private ExecutorService executor = Executors.newFixedThreadPool(10);

    @Inject
    private KafkaTemplate<String, String> kafkaTemplate;


    /**
     * 多线程发送消息到 Kafka
     */
    public void sendMessage(String topic, String message) {
        executor.execute(() -> syncSendMessage(topic, message));
        System.out.println("发送kafka消息成功");
        log.debug("send message='{}' to topic='{}' ", message, topic); // 打印日志会影响发送效率
    }

    /**
     * 同步发送消息
     */
    private void syncSendMessage(String topic, String message) {
        kafkaTemplate.send(topic,"demo", message);
    }

    /**
     * 异步发送消息
     */
    private void asyncSendMessage(String topic, String message) {
        // the KafkaTemplate provides asynchronous send methods returning a Future
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, message);

        // register a callback with the listener to receive the result of the send asynchronously
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {

            @Override
            public void onSuccess(SendResult<String, String> result) {
                log.debug("SUCCESS: send message='{}' to topic='{}' with offset={}", message, topic, result.getRecordMetadata().offset());
            }

            @Override
            public void onFailure(Throwable ex) {
                log.error("ERROR: unable to send message='{}' to topic='{}'", message, topic, ex);
            }
        });
    }

}